package com.navi.loaddata.job;

import com.navi.loaddata.core.etlException.ETLException2;
import com.navi.loaddata.core.mysqlEntity.EtlConfD;
import com.navi.loaddata.core.mysqlEntity.EtlConfDDefect;
import com.navi.loaddata.core.mysqlEntity.EtlConfDMapper;
import com.navi.loaddata.core.mysqlEntity.EtlConfDModel;
import com.navi.loaddata.core.util.LoggerUtil;
import com.navi.loaddata.email.MailService;
import com.navi.loaddata.loadService.IEtlConfServices2;
import com.navi.loaddata.loadService.impl.DefectData.LoadDefectDataBean;
import com.navi.loaddata.loadService.impl.DefectData.LoadDefectData_CService;
import org.quartz.Job;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.StringUtils;

import java.sql.Timestamp;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;

public class LoadDefectData_C_His implements Job {

    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    @Autowired
    private IEtlConfServices2 etlConfServices;

    @Autowired
    private LoadDefectData_CService LoadDefectData_CService;

    @Autowired
    private LoadDefectDataBean loadDefectDataBean;


    @Override
    public void execute(JobExecutionContext context) throws JobExecutionException {
        //get JOB
        EtlConfD etlConfD = etlConfServices.getNewConf(context);
        if (null == etlConfD) return;

        List<EtlConfDDefect> etlCurJobList = loadDefectDataBean.getCurJob(etlConfD);

        try {
            //若current job的runEndTimeStamp与nowTime时间差小于BufferTime+QueryTime，则his job不运行
            for (EtlConfDDefect etlCurJob : etlCurJobList) {
                if (etlCurJob.getRun_end_timestamp() != null) {
                    Timestamp nowTime = Timestamp.valueOf(LocalDateTime.now());
                    long duration = (nowTime.getTime() - etlCurJob.getRun_end_timestamp().getTime()) / 1000 / 60;
                    if ((int) duration >= (etlCurJob.getBuffer_minute() + etlCurJob.getQuery_minute())) {
                        etlConfD.setExt2("N");
                    } else {
                        etlConfD.setExt2("Y");
                    }
                } else {
                    etlConfD.setExt2("N");
                }
                loadDefectDataBean.updateEtlConfD(etlConfD);
                if ("N".equals(etlConfD.getExt2())) {
                    return;
                }
            }

            //判断如果从etl_conf_d表中查询对应his job的ext_2值为Y，则不做后续处理
            if (etlCurJobList.size() > 0) {
                for (EtlConfDDefect etlCurJob : etlCurJobList) {
                    if ("Y".equals(etlCurJob.getExt_2())) {
                        etlConfD.setExt2("N");
                    } else {
                        etlConfD.setExt2("Y");
                    }
                }
            } else {
                etlConfD.setExt2("Y");
            }
            loadDefectDataBean.updateEtlConfD(etlConfD);
            if ("N".equals(etlConfD.getExt2())) {
                return;
            }

            etlConfD = etlConfServices.beginJob(etlConfD);
            if (StringUtils.isEmpty(etlConfD.getQueryMinute())) return;
            startETLByOpes(etlConfD);
        } catch (Exception e) {
            LoggerUtil.stackTraceToString(e);
        }
    }

    private void startETLByOpes(EtlConfD etlConfD) {

        //获取站点
        EtlConfDModel etlConfDModel = EtlConfDMapper.INSTANCE.etlConfDToModel(etlConfD);

        List<String> errorOpeList = new ArrayList<>();

//        List<String> opeList = LoadDefectData_CService.getOpeList(etlConfDModel);
//        if(opeList.isEmpty()) return;//查询Opelist异常
        List<String> opeList = etlConfServices.getOpesList(etlConfD);

        if (null==opeList.get(0)){//没有数据
            String[] resMsg ={"0","0",null};//okCnt,errCnt,errMsg
            EtlConfDModel etlConfDByOpe = etlConfServices.endJobOfModel(etlConfD);
            etlConfServices.addHisOfModel(etlConfDByOpe, resMsg);
            etlConfServices.endJob(etlConfD);
            return;
        }

        long loopCnt = 0;
        //循环OPE
        for (String opeid : opeList) {
            etlConfDModel.setOpe(opeid);//跑单站点
            String[] resMsg = new String[3];//okCnt,errCnt,errMsg
            try {
                resMsg = LoadDefectData_CService.loadDefectDataMain(etlConfDModel, etlConfD);
                EtlConfDModel etlConfDByOpe = etlConfServices.endJobOfModel(etlConfD);
                if (null != resMsg[2] && resMsg[2].indexOf("deadlock") > -1) return;
                if (null == etlConfDByOpe) return;
                etlConfDByOpe.setOpe(opeid);
                etlConfServices.addHisOfModel(etlConfDByOpe, resMsg);
            } catch (ETLException2 e) {
                String err = e.getErrMsg();
                //print Log
                logger.error("Job {} [{}]-[{}] OPE={} FAIL TO SYNCHRONIZE.\n{}\n{}",
                        etlConfD.getProjectName() + "_" + etlConfD.getJobGroupName() + "_" + etlConfD.getJobName(),
                        etlConfD.getRunStartTimestamp(), etlConfD.getRunEndTimestamp(),opeid,
                        e.getErrorCode().getErrorMsg(), LoggerUtil.stackTraceToString(e)
                );

                errorCatch(etlConfD, loopCnt, opeid, errorOpeList, LoggerUtil.stackTraceToString(e));

//                EtlConfDModel etlConfDByOpe = etlConfServices.endJobOfModel(etlConfD);
//                if (null == etlConfDByOpe) return;
//                etlConfDByOpe.setOpe(opeid);
//                resMsg[0] = "0";
//                resMsg[1] = "-1";
//                resMsg[2] = err;
//                etlConfServices.addHisOfModel(etlConfDByOpe, resMsg);
//
//                if (99999 == e.getErrorCode().getErrorCode()) {//无法连接,SLEEP
//                    etlConfServices.resetJob(etlConfD);
//                    return;
//                }
                return;
            } catch (Exception e) {
                String err = LoggerUtil.stackTraceToString(e);
                logger.error("Job {} [{}]-[{}] OPE={} FAIL TO SYNCHRONIZE.\n{}",
                        etlConfD.getProjectName() + "_" + etlConfD.getJobGroupName() + "_" + etlConfD.getJobName(),
                        etlConfD.getRunStartTimestamp(), etlConfD.getRunEndTimestamp(),opeid,
                        LoggerUtil.stackTraceToString(e)
                );

                errorCatch(etlConfD, loopCnt, opeid, errorOpeList, LoggerUtil.stackTraceToString(e));

//                EtlConfDModel etlConfDByOpe = etlConfServices.endJobOfModel(etlConfD);
//                if (null == etlConfDByOpe) return;
//                etlConfDByOpe.setOpe(opeid);
//                resMsg[0] = "0";
//                resMsg[1] = "-1";
//                resMsg[2] = err;
//                etlConfServices.addHisOfModel(etlConfDByOpe, resMsg);
//
//                if (null != e && "Y".equals(etlConfServices.specialErr(etlConfDByOpe.getJobName(), err))) {//无法连接,SLEEP
//                    etlConfServices.resetJob(etlConfD);
//                    return;
//                }
                return;
            }
        }
        etlConfServices.endJob(etlConfD);
//        etlConfD.setExt2("N");
//        loadDefectDataBean.updateEtlConfD(etlConfD);

        //跑current job，若runEndTimeStamp与nowTime的时间差大于BufferTime，将Ext_2置成Y，供His job做参考
        List<EtlConfDDefect> etlCurJobList = loadDefectDataBean.getCurJob(etlConfD);

        if (etlCurJobList.size() > 0) {
            for (EtlConfDDefect etlCurJob : etlCurJobList) {
                if (etlCurJob.getRun_end_timestamp() != null) {
                    Timestamp nowTime = Timestamp.valueOf(LocalDateTime.now());
                    long duration = (nowTime.getTime() - etlCurJob.getRun_end_timestamp().getTime()) / 1000 / 60;
                    if ((int) duration >= (etlCurJob.getBuffer_minute() + etlCurJob.getQuery_minute())) {
                        etlConfD.setExt2("N");
                    } else {
                        etlConfD.setExt2("Y");
                    }
                } else {
                    etlConfD.setExt2("N");
                }
                loadDefectDataBean.updateEtlConfD(etlConfD);
            }
        }

        //His Job跑到final_end_timestamp的时间，则ext_2置成N
        if (etlConfD.getFinalEndTimestamp().equals(etlConfD.getRunEndTimestamp())) {
            etlConfD.setExt2("N");
            loadDefectDataBean.updateEtlConfD(etlConfD);
        }
        logger.info("loadDefectData_C run completed!");
    }


    private void errorCatch(EtlConfD etlConfD, long loopCnt, String opeid, List<String> errorOpeList, String errorLog) {
        //获取加入到opeList中的重复站点数量
        loopCnt = errorOpeList.stream().filter(s -> s.equals(opeid)).count();

        //报错站点数据重跑
        errorOpeList.add(opeid);

        if (loopCnt <= 4) {
            //获取站点
            EtlConfDModel etlConfDModel = EtlConfDMapper.INSTANCE.etlConfDToModel(etlConfD);

            etlConfDModel.setOpe(opeid);//跑单站点
            String[] resMsg = new String[3];//okCnt,errCnt,errMsg
            try {
                resMsg = LoadDefectData_CService.loadDefectDataMain(etlConfDModel, etlConfD);
                EtlConfDModel etlConfDByOpe = etlConfServices.endJobOfModel(etlConfD);
                if (null != resMsg[2] && resMsg[2].indexOf("deadlock") > -1) return;
                if (null == etlConfDByOpe) return;
                etlConfDByOpe.setOpe(opeid);
                etlConfServices.addHisOfModel(etlConfDByOpe, resMsg);
            } catch (ETLException2 e) {
                String err = e.getErrMsg();
                //print Log
                logger.error("Job {} [{}]-[{}] OPE={} FAIL TO SYNCHRONIZE.\n{}\n{}",
                        etlConfD.getProjectName() + "_" + etlConfD.getJobGroupName() + "_" + etlConfD.getJobName(),
                        etlConfD.getRunStartTimestamp(), etlConfD.getRunEndTimestamp(),opeid,
                        e.getErrorCode().getErrorMsg(), LoggerUtil.stackTraceToString(e)
                );

                errorCatch(etlConfD, loopCnt, opeid, errorOpeList, LoggerUtil.stackTraceToString(e));

//                EtlConfDModel etlConfDByOpe = etlConfServices.endJobOfModel(etlConfD);
//                if (null == etlConfDByOpe) return;
//                etlConfDByOpe.setOpe(opeid);
//                resMsg[0] = "0";
//                resMsg[1] = "-1";
//                resMsg[2] = err;
//                etlConfServices.addHisOfModel(etlConfDByOpe, resMsg);
//
//                if (99999 == e.getErrorCode().getErrorCode()) {//无法连接,SLEEP
//                    etlConfServices.resetJob(etlConfD);
//                    return;
//                }
                return;
            } catch (Exception e) {
                String err = LoggerUtil.stackTraceToString(e);
                logger.error("Job {} [{}]-[{}] OPE={} FAIL TO SYNCHRONIZE.\n{}",
                        etlConfD.getProjectName() + "_" + etlConfD.getJobGroupName() + "_" + etlConfD.getJobName(),
                        etlConfD.getRunStartTimestamp(), etlConfD.getRunEndTimestamp(),opeid,
                        LoggerUtil.stackTraceToString(e)
                );

                errorCatch(etlConfD, loopCnt, opeid, errorOpeList, LoggerUtil.stackTraceToString(e));

//                EtlConfDModel etlConfDByOpe = etlConfServices.endJobOfModel(etlConfD);
//                if (null == etlConfDByOpe) return;
//                etlConfDByOpe.setOpe(opeid);
//                resMsg[0] = "0";
//                resMsg[1] = "-1";
//                resMsg[2] = err;
//                etlConfServices.addHisOfModel(etlConfDByOpe, resMsg);
//
//                if (null != e && "Y".equals(etlConfServices.specialErr(etlConfDByOpe.getJobName(), err))) {//无法连接,SLEEP
//                    etlConfServices.resetJob(etlConfD);
//                    return;
//                }
                return;
            }
        }else if (loopCnt > 4) {
            try {
                MailService mail = new MailService();
                mail.sendHtmlMail("chengyang.fu@s-navimentum.com", "DefectSummary同一站点报错超过4次", etlConfD.getProjectName() + "_" + etlConfD.getJobGroupName() + "_" + etlConfD.getJobName()
                        + "_" + etlConfD.getRunStartTimestamp() + "_" + etlConfD.getRunEndTimestamp() + "_" + opeid + "_" +
                        errorLog);
            } catch (Exception ex) {
                LoggerUtil.stackTraceToString(ex);
            }
        }
    }
}
